package com.qf.safe;

import com.qf.util.ConnectionUtil;
import com.rabbitmq.client.*;
import com.rabbitmq.client.impl.AMQBasicProperties;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Publisher {

    @Test
    public void publish() throws Exception {
        //1. connection
        Connection connection = ConnectionUtil.getConnection();

        //2. channel
        Channel channel = connection.createChannel();

        //3. 开启confirm机制
        channel.confirmSelect();

        //4. 添加confirm机制的异步回调
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                // 代表消息已经送达到Exchange
                System.out.println("代表消息已经送达到Exchange");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                // 代表消息在传输的过程中丢了.
                System.out.println("代表消息在传输的过程中丢了.");
            }
        });

        //5. 添加return机制的异步回调
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("发送到Exchange上的消息没有路由到队列,消息是: " + new String(body));
            }
        });

        //6. 准备消息的其他属性
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder().deliveryMode(2).build();

        //7. 发送消息到, 开启Return机制,需要将mandatory设置为true
        channel.basicPublish("fanout_exchange","",true,props,"我是消息!".getBytes());

        //8. 避免服务停止
        System.in.read();
    }

}
